
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <errno.h>
#include <malloc.h>
#include <string.h>
#include <pthread.h>
#include <assert.h>
#include "Facade.h"
#include "NetUtils.h"
#include "Handler.h"
#include "Debug.h"
#include "aclHandler.h"
/*
 * Worker thread: main event loop
 */
void * Thread_Worker_Start(void *arg) 
{
    LIBEVENT_THREAD *me = (LIBEVENT_THREAD *) arg;
    Facade *pR = Facade::Instance();
    /* Any per-thread setup can happen here; thread_init() will block until
     * all threads have finished initializing.
     */
    pR->GetThreads()->InitCountIncrease();
    return (void*) event_base_loop(me->base, 0);
}
/*
 * Processes an incoming "handle a new connection" item. This is called when
 * input arrives on the libevent wakeup pipe.
 */
void Thread_OnRead_Process(int fd, short which, void *arg) 
{
 
	LIBEVENT_THREAD *me = (LIBEVENT_THREAD *) arg;
	CQ_ITEM *item;
	char buf[INET6_ADDRSTRLEN + 1];
        int rlen = read(fd, buf, INET6_ADDRSTRLEN);
        if (rlen  < 1)
        {
             fprintf(stderr, "Can't read from libevent pipe\n");
             return;
	}
	else
		buf[rlen] = '\0';
	DEBUG2("signal length = ", rlen);
	assert((rlen == 1)||rlen > 6);

	Facade *pR = Facade::Instance();
	if(strncmp( buf, NANTX_EVENT_TIMER, 1)  == 0)
	{
		item = pR->GetThreads()->cq_peek(&me->new_conn_queue);
		if(NULL != item)
		{
			pR->DoTimer();
			pR->GetThreads()->cqi_free(item);
		}   
		return;
	}

	item = pR->GetThreads()->cq_peek(&me->new_conn_queue);
	if (NULL != item) {
		conn *c = pR->GetNetWork()->conn_new(item->sfd, item->init_state, item->event_flags, item->read_buffer_size,  me->base);
		if (c == NULL) {
				if (pR->GetSettings()->verbose > 0) {
					fprintf(stderr, "Can't listen for events on fd %d\n",
							item->sfd);
				}
				close(item->sfd);
		}
		if(rlen > 6){
			bcopy(buf, c->cIP, rlen);
                        c->cIP[INET6_ADDRSTRLEN] = '\0';
                       DEBUG2("--------- NEW Connection, IP = ", c->cIP);
                }
		pR->GetThreads()->cqi_free(item);
	}
}

/*
 * Sets a connection's current state in the state machine. Any special
 * processing that needs to happen on certain state transitions can
 * happen here.
 */
void Conn_Set_State(conn *c, int state) {


	DEBUG6("threadID=", pthread_self()," set conn(", c->sfd,")'s state = ", state);
	DEBUG4("conn_closing=",conn_closing, "; conn_read = ",conn_read );
	assert(c != NULL);
	if (state != c->state) 
		c->state = state;
}

bool Update_Event(conn *c, const int new_flags) {
	assert(c != NULL);
	struct event_base *base = c->event.ev_base;
	if (c->ev_flags == new_flags)
		return true;
	if (event_del(&c->event) == -1)
        {
                DEBUG1("false to event_del");
		return false;
	} 

	DEBUG1("success to event_del");
        assert(c->sfd > 0);
            
	event_set(&c->event, c->sfd, new_flags, Thread_Event_Handler, (void *)c);
	event_base_set(base, &c->event);
	c->ev_flags = new_flags;
	if (event_add(&c->event, 0) == -1)
        {
                DEBUG1("event_add false");
		return false;
	}
	DEBUG1("event_add success");
	return true;
}

/*
 * Sets whether we are listening for new connections or not.
 */
void Accept_New_Connection(const bool do_accept) {
	//conn *next;
	Facade *pR = Facade::Instance();
	if (! pR->GetThreads()->is_listen_thread())
		return;
       pR->GetThreads()->conn_listen_accept( do_accept );
}

void Handler_Dispatch(conn *c) {
	bool stop = false;
	int sfd = 0;

	//int sendbuf = 4*1024*1024;
	//struct timeval timeout = {0,100}; 
	int flags = 1;
	// struct sockaddr_storage addr;
	struct sockaddr_in addr;
	socklen_t addrlen = sizeof( addr );
	char sip[ INET6_ADDRSTRLEN ];
	int iread = 0;
	Facade *pR = Facade::Instance();
	aclHandler *pAcl = aclHandler::Instance();
	assert(c != NULL);
	while (!stop) {
		switch(c->state) {
			case conn_listening:
				addrlen = sizeof(addr);
				if ((sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen)) == -1) {
					if (errno == EAGAIN || errno == EWOULDBLOCK) {
						/* these are transient, so don't log anything */
						stop = true;
					} else if (errno == EMFILE) {
						cerr<<"Too many connection ..."<< endl;
						///////////////////////////////////////////////
						//
						//  Accept_New_Connection(false);
						//
						//////////////////////////////////////////////  
                                                close(c->sfd);   
						stop = true;
					} else {
						perror("accept()");
						ERRO<<"accept error "<<endl;
//////////////////////////
                                                close(c->sfd);
/////////////////////////////////////////////  ???????????????????????????      
						stop = true;
                                           
					}
					break;
				}

				if(addr.sin_family != AF_INET){
					// strcpy(sip, inet_ntoa(((struct sockaddr_in6 *)(&addr))->sin6_addr));
					strcpy (sip, "127.0.0.1");
					//is  ipv6, pass
				}
				else
					strcpy(sip, inet_ntoa(addr.sin_addr));

				/*
				   struct sockaddr_storage *stoaddr;
				   stoaddr = (struct sockaddr_storage *)(&addr);
				   switch ( stoaddr->ss_family )
				   {
				   case AF_INET:
				   DEBUG1("using AF_INET !!!");
				   strcpy(sip, inet_ntoa(addr.sin_addr));
				   break;
				   case AF_UNIX:
				   default:
				   strcpy (sip, "127.0.0.1");
				   break;
				   }
				 */
				DEBUG2("client try to connect, ip = ", sip);
				if(pAcl != NULL && !pAcl->AclCheck( sip))
				{
					close(sfd);
					ERRO<< "ip not in ACL list,can't accept it : "<< sip <<endl;
					stop = true;
					break;
				}

				if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
					perror("setting O_NONBLOCK");
					close(sfd);
					break;
				}

				pR->GetThreads()->dispatch_conn_new(sfd, conn_read, EV_READ | EV_PERSIST, DATA_BUFFER_SIZE, sip);
				break;

			case conn_read:
				iread = pR->GetNetWork()->read_network(c); 
				if( iread > 0) 
					// if ( pR->GetNetWork()->try_read_network(c) > 0) 
				{
					DEBUG2("read rbytes=", c->rbytes);
					// printf("%s:\n%s\n", __func__, c->rcurr);
					int rets = pR->GetHandler()->ReadProcess(c);
					DEBUG4("after ReadProcess, set state = ", rets, ";  conn_read = ", conn_read);
					Conn_Set_State(c, rets);
					if( rets == conn_closing)
						break;
				}
				if(iread < 0)
				{
					Conn_Set_State(c, conn_closing);
					break;
				}
				/* we have no command line and no data to read from network */
				if (!Update_Event(c, EV_READ | EV_PERSIST)) {
					DEBUG1("update event");
					Conn_Set_State(c, conn_closing);
					break;
				}
				DEBUG1("exit while condition and Set Event EV_READ to waitting for read! ");  
				stop = true;
				break;
			case conn_write:
				DEBUG1("receive write EVENT!");
				if (c->wbytes > 0 ) 
				{
					if (pR->GetHandler()->WriteProcess(c) == conn_closing) 
					{
						Conn_Set_State(c, conn_closing);
						break;
					}
				} 
				/* fall through... */
			case conn_mwrite:
				break;
			case conn_closing:
                                DEBUG1("Yes, close conn! ");
				pR->GetNetWork()->conn_close(c);
				stop = true;
				break;
		}
	}

	return;
}

void Thread_Event_Handler(const int fd, const short which, void *arg) {
	Facade *pR = Facade::Instance();
	conn *c;
    c = (conn *)arg;
    assert(c != NULL);
    c->which = which;
    /* sanity */
    if (fd != c->sfd) {
        pR->GetNetWork()->conn_close(c);
        return;
    }
     
    Handler_Dispatch(c);
    /* wait for next event */
    return;
}

